package cn.spark.study.core;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;
import scala.tools.nsc.util.SourcePath;

import java.util.List;

/**
 * 取最大的前三个数字
 */
public class Top3 {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("top3").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        String path = "E:\\BaiduNetdiskDownload\\Spark从入门到精通（Scala编程、案例实战、高级特性、Spark内核源码剖析、Hadoop高端）\\第40讲-Spark核心编程：高级编程之topn\\文档\\top.txt";
        JavaRDD<String> lines = sc.textFile(path);
        JavaPairRDD<Integer, String> pairs = lines.mapToPair(new PairFunction<String, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(String line) throws Exception {
                return new Tuple2<Integer, String>(Integer.parseInt(line), line);
            }
        });

        JavaPairRDD<Integer, String> sortedPairs = pairs.sortByKey(false);

        JavaRDD<Integer> sortedNumbers = sortedPairs.map(new Function<Tuple2<Integer, String>, Integer>() {
            @Override
            public Integer call(Tuple2<Integer, String> t) throws Exception {
                return t._1;
            }
        });

        List<Integer> sortedNumberList = sortedNumbers.take(3);
        for (Integer integer : sortedNumberList) {
            System.out.println(integer);
        }

        sc.close();
    }

}
